package com;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @desc: Oracle 表同步到 MySQL
 * @author 惊云
 * @date 2023/8/13 16:38
 */
public class OracleStreamToMySQLExample {

    public static void main(String[] args) {
        //1.获取stream的执行环境
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        senv.setParallelism(1);
        senv.disableOperatorChaining();
        //2.创建表执行环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(senv);

        String sourceTable = "CREATE TABLE oracle_cdc_PROJECT_TYPE_INFO " +
                "( ID CHAR, " +
                "PROJECT_TYPE VARCHAR, " +
                "PROJECT_TYPE_NAME VARCHAR, " +
                "PROJECT_SUB_TYPE VARCHAR,"+
                "PROJECT_SUB_TYPE_NAME VARCHAR,"+
                "FUNCTION_SCOPE VARCHAR,"+

                "PRIMARY KEY(ID) NOT ENFORCED) WITH (\n" +
                "'connector' = 'oracle-cdc',\n" +
                "'hostname' = '10.7.204.250',\n" +
                "'port' = '1521',\n" +
                "'username' = 'pomp',\n" +
                "'password' = 'pomp#pre##',\n" +
                "'database-name' = 'pomppre',\n" +//select name from v$database;
                "'schema-name' = 'pomp',\n" +//select SYS_CONTEXT('USERENV','CURRENT_SCHEMA') CURRENT_SCHEMA from dual;
                "'debezium.snapshot.mode' = 'schema_only',\n" +
                //snapshot.mode = initial 快照包括捕获表的结构和数据。指定此值将用捕获表中数据的完整表示填充主题。
                //snapshot.mode = schema_only 快照只包含捕获表的结构。如果希望连接器仅捕获快照之后发生的更改的数据，请指定此值。
                "'scan.incremental.snapshot.enabled' = 'true',\n" +
                //scan.incremental.snapshot.enabled 增量快照是一种读取表快照的新机制。增量快照与旧的快照机制相比有很多优点，包括:
                // (1)在快照读取期间源可以并行;(2)在快照读取期间源可以在块粒度上执行检查点;(3)在快照读取之前源不需要获取ROW SHARE MODE锁。
                "'scan.incremental.snapshot.chunk.size' = '8096' ,\n" +
                //表快照的块大小(行数)，当读取表快照时，捕获的表被分割成多个块。
                "'scan.snapshot.fetch.size' = '1024',\n" +
                //读取表快照时每个轮询的最大读取大小。
                "'connect.max-retries' = '3',\n" +
                //连接器应该重试构建Oracle数据库服务器连接的最大重试次数。
                "'connection.pool.size'= '20',\n" +
                //连接池大小
                "'debezium.log.mining.strategy' = 'online_catalog',\n" +
                //online_catalog -使用数据库的当前数据字典来解析对象id，并且不向在线重做日志中写入任何额外的信息。
                // 这使得LogMiner的挖掘速度大大提高，但代价是无法跟踪DDL的变化。如果捕获的表模式很少或从不更改，那么这是理想的选择。
                "'debezium.log.mining.archive.destination.name' = 'log_archive_dest_1',\n" +
                "'debezium.log.mining.continuous.mine'='true'," +
                "  'table-name' = 'USER_INFO'\n" +
                ")";
        tEnv.executeSql(sourceTable);

        //加上打印后，虽然可以实时看到增删改查记录，但是这些后续操作并不会插入到目标表。如果不加这句打印，则程序无问题
        tEnv.executeSql("select * from oracle_cdc_source").print();

        String sinkTable = "CREATE TABLE mysql_cdc_sink (" +
                "  ID INT,\n" +
                "  USERNAME STRING,\n" +
                "  PASSWORD STRING,\n" +
                "PRIMARY KEY(ID) NOT ENFORCED\n" +
                ") WITH (\n" +
                "'connector' = 'jdbc',\n" +
                "'driver' = 'com.mysql.cj.jdbc.Driver',\n" +
                "'url' = 'jdbc:mysql://localhost:3306/test?rewriteBatchedStatements=true',\n" +
                "'username' = 'flink_cdc_user',\n" +
                "'password' = 'flink@cdc',\n"+
                "  'table-name' = 'user_new',\n" +
                "  'connection.max-retry-timeout' = '60s'\n" +
                ")";
        tEnv.executeSql(sinkTable);

        tEnv.executeSql("insert into mysql_cdc_sink select ID,USERNAME,PASSWORD from oracle_cdc_source");
    }



}
